package com.example.Utils;

import com.example.entity.StockTrade;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;

import java.io.Serializable;

@Slf4j
public class StockTradeMapper implements MapFunction<String, StockTrade>, Serializable {
    private static final long serialVersionUID = 1L;

    // 将 ObjectMapper 标记为 transient
    private transient ObjectMapper objectMapper;

    private ObjectMapper getObjectMapper() {
        if (objectMapper == null) {
            objectMapper = new ObjectMapper();
        }
        return objectMapper;
    }

    @Override
    public StockTrade map(String value) throws Exception {
        if (value == null || value.trim().isEmpty()) {
            log.warn("Received empty or null message: {}", value);
            return null; // 返回 null 或者抛出自定义异常
        }

        try {
            StockTrade trade = getObjectMapper().readValue(value, StockTrade.class);
            return trade;
        } catch (Exception e) {
            log.error("Error mapping stock trade: {}", value, e);
            return null; // 返回 null 或者抛出自定义异常
        }
    }
}
